package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.processor.TaskId;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * 任务处理的公共业务
 */
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;
    @Autowired
    private CacheService cacheService;

    @Scheduled(cron = "0 */5 * * * ?") // 一部分数据已经到了时间 该执行了
    @PostConstruct // 这个bean生命周期的初始化方法
    public void reloadData() {
        clearCache();
        log.info("数据库数据同步到缓存");
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);

        //查看小于未来5分钟的所有任务
        List<Taskinfo> allTasks = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime,calendar.getTime()));

        if(allTasks != null && allTasks.size() > 0){
            for (Taskinfo taskinfo : allTasks) {
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo,task);
                task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                addTaskToCache(task);
            }
        }
    }

    private void clearCache(){
        // 删除缓存中未来数据集合和当前消费者队列的所有key
        Set<String> futurekeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_
        Set<String> topickeys = cacheService.scan(ScheduleConstants.TOPIC + "*");// topic_
        cacheService.delete(futurekeys);
        cacheService.delete(topickeys);
    }

    /**
     * @Scheduled
     * 每分钟的第0秒
     */
    @Scheduled(cron = "0 */1 * * * ?")
    public void refresh() {

        String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if(StringUtils.isNotBlank(token)){

            System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");

            // 获取所有未来数据集合的key值
            Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*
            for (String futureKey : futureKeys) { // future_250_250

                String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
                //获取该组key下当前需要消费的任务数据
                Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                if (!tasks.isEmpty()) {
                    //将这些任务数据添加到消费者队列中
                    cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
                    System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
                }
            }
        }

    }

    /**
     * 消费任务
     * @param type
     * @param priority
     * @return
     */
    @Override
    public Task poll(int type, int priority) {
        Task task = null;
        // 1. 从对列中获取任务 - （获取并删除）
        try {
            String key = ScheduleConstants.TOPIC+type+"_"+ priority;
            String task_josn = cacheService.lRightPop(key);

            // 2. 从数据库删除
            if(StringUtils.isNotBlank(task_josn)){
                task = JSON.parseObject(task_josn, Task.class);//不然会空指针
                updateDb(task.getTaskId(), ScheduleConstants.EXECUTED);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("消费失败 消费异常");
        }

        return task;
    }

    /**
     * 取消任务
     * @param taskId
     * @return   false  删除成功
     */
    @Override
    public boolean cancelTask(long taskId) {
        // 1. 从数据库中删除数据
        Task task = updateDb(taskId,ScheduleConstants.CANCELLED);

        // 2. 从缓存中删除数据
        // 根据key  和value去删除缓存
        // 2.1定义key
        String key = task.getTaskType()+"_"+ task.getPriority();
        key = ScheduleConstants.TOPIC + key;
        cacheService.lRemove(key,0,JSON.toJSONString(task));// 从消费队列删除

        key = ScheduleConstants.FUTURE + key;
        cacheService.zRemoveRangeByScore(key,task.getExecuteTime(),task.getExecuteTime());// 从未来队列删

        return false;
    }

    /**
     * 删除 任务， 并且 更新日志
     * @param taskId 任务ID
     * @param status 任务的状态
     * @return 返回值是删除的任务
     */
    private Task updateDb(long taskId,Integer status) {
        //1.1 删除Db任务
        taskinfoMapper.deleteById(taskId);
        //1.2 修改日志
        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
        taskinfoLogs.setStatus(status);
        taskinfoLogsMapper.updateById(taskinfoLogs);

        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs,task);
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        return task;
    }

    /**
     * 添加任务
     * @param task
     * @return
     */
    @Override
    public long addTask(Task task) {

        // 1. 添加任务到数据库
        boolean taskToDb = addTaskToDb(task);

        // 2. 添加到缓存中
        if(taskToDb){

            addTaskToCache(task);
        }
        return task.getTaskId();
    }

    /**
     * 添加任务到缓存
     * @param task
     */
    private void addTaskToCache(Task task) {
        // 定义key
        String key = task.getTaskType()+"_"+ task.getPriority();

        // 获取这个任务的执行是时间
        //获取5分钟之后的时间  毫秒值
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        long nextScheduleTime = calendar.getTimeInMillis();
        // 执行的时间戳
        long executeTime = task.getExecuteTime();

        if(executeTime < System.currentTimeMillis()){
            // 队列的名字
            key = ScheduleConstants.TOPIC + key;
            cacheService.lLeftPush(key, JSON.toJSONString(task));// list - 立即消费
        }else if(executeTime<nextScheduleTime){
            // 队列的名字
            key = ScheduleConstants.FUTURE + key;
            cacheService.zAdd(key,JSON.toJSONString(task), task.getExecuteTime());// zset - 未来消费对象
        }
    }

    /**
     * 添加任务到数据库
     * @param task
     * @return
     */
    private boolean addTaskToDb(Task task) {

        boolean flag = true;

        try {
            // 1.1 taskInfo- 任务基本信息
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskinfoMapper.insert(taskinfo);

            // 把生成的ID赋值给task
            task.setTaskId(taskinfo.getTaskId());

            // 1.2 taskInfoLog - 乐观锁更新
            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo,taskinfoLogs);
            taskinfoLogs.setVersion(1);//第一次新增 1
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskinfoLogsMapper.insert(taskinfoLogs);
        } catch (BeansException e) {
            e.printStackTrace();
            flag = false;
        }

        return flag;
    }
}
